package com.project.RocketRoute.executor;

import com.project.RocketRoute.model.RocketRouteBean;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;

@Component(value = "RocketMqListenerConsumer")
public class RocketMqListenerConsumer implements MessageListenerConcurrently {

    @Autowired
    private RocketMqProductExecutor rocketMqProduct;

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        try
        {
            for(MessageExt msg : list){
                String name = String.format("%s.%s", msg.getTopic(), msg.getTags());
                List<RocketRouteBean> bean = RocketRouteSetting.getBean(name);
                for (RocketRouteBean item : bean){
                    String in_str = new String(msg.getBody());
                    System.out.println("===============>listener message: " + in_str);
                    if (item.isAsync()){
                        item.syncInvoke(in_str).subscribe((json) -> rocketMqProduct.sendGroup(item.getRouteMethod().groupId(), item.getRouteMethod().topic(), item.getResult(), json));
                    } else {
                        rocketMqProduct.sendGroup(item.getRouteMethod().groupId(), item.getRouteMethod().topic(), item.getResult(), item.invoke(in_str));
                    }
                    System.out.println("===============>end listener message");
                }
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        } catch (Exception e){
            e.printStackTrace();
        }

        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
    }


}
